package com.ww.microrpc.netty;

import com.ww.microrpc.exception.RpcServerException;
import com.ww.microrpc.executer.MethodExecutors;
import com.ww.microrpc.netty.handler.ProtocolFrameDecoderHandler;
import com.ww.microrpc.netty.handler.RpcDecoderHandler;
import com.ww.microrpc.netty.handler.RpcEncoderHandler;
import com.ww.microrpc.netty.handler.server.*;
import com.ww.microrpc.util.ThreadPoolUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

public class NettyServer {
    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private Thread thread;
    private MethodExecutors methodExecutors;
    private final EventLoopGroup eventLoopGroupSelector;
    private final EventLoopGroup eventLoopGroupBoss;
    private NettyServerConfig nettyServerConfig;
    private ServerBootstrap serverBootstrap;
    private Map<String, Object> serviceMap = new HashMap<>();

    public NettyServer(MethodExecutors methodExecutors, NettyServerConfig nettyServerConfig) {
        this.methodExecutors = methodExecutors;
        this.nettyServerConfig = nettyServerConfig;
        this.serverBootstrap = new ServerBootstrap();

        if (useEpoll()) {
            this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyEPOLLBoss_%d", this.threadIndex.incrementAndGet()));
                }
            });

            this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                }
            });
        } else {
            this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
                }
            });

            this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
                }
            });
        }
    }

    @PostConstruct
    public void init(){
        if (nettyServerConfig.getEnabled()) {
            this.start();
        }
    }

    public void start() {
        ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.createThreadPool(NettyServer.class.getSimpleName(), 16, 32);

        ServerBootstrap bootstrap = this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
//                    .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getServerSocketBacklog())
//                    .option(ChannelOption.SO_REUSEADDR, true)
//                    .option(ChannelOption.SO_KEEPALIVE, false)
//                    .childOption(ChannelOption.TCP_NODELAY, true)
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getIp(), this.nettyServerConfig.getPort()))
                .childHandler(new ChannelInitializer<>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addLast(new ProtocolFrameDecoderHandler());
                        channel.pipeline().addLast(new RpcDecoderHandler());
                        channel.pipeline().addLast(new RpcEncoderHandler());
                        // idle event when reader timeout
                        // idle event when write timeout
                        // idle event when reader or write timeout
                        channel.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getClientTimeout(), 0, 0));
                        channel.pipeline().addLast(new ChannelDuplexHandler() {
                            @Override
                            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                                IdleStateEvent event = (IdleStateEvent) evt;
                                if (event.state() == IdleState.READER_IDLE) {
                                    logger.info("I haven't sent a message in a long time: {}, this channel has be closed: {} ", nettyServerConfig.getClientTimeout(), ctx.channel().id());
                                    ctx.channel().close();
                                }
                            }
                        });
                        channel.pipeline().addLast(new RpcServerHandler(threadPoolExecutor, methodExecutors));
                        channel.pipeline().addLast(new RequestMessageHandler(threadPoolExecutor, methodExecutors));
                        channel.pipeline().addLast(new PingMessageHandler());
                        channel.pipeline().addLast(new PongMessageHandler());
                        channel.pipeline().addLast(new ErrorMessageHandler());
                        channel.pipeline().addLast(new NettyConnectManageHandler());
                    }
                });
//                  serverBootstrap.option(ChannelOption.SO_BACKLOG, 128)
//                  serverBootstrap.option(ChannelOption.SO_RCVBUF, 1024 * 1000 * 1000)
//                  serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(1200 * 10));
//                  serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(4096, 8192, 65535))
//                  serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);

//            if (this.nettyServerConfig.getServerSocketSndBufSize() > 0) {
//                logger.info("server set SO_SNDBUF to {}", this.nettyServerConfig.getServerSocketSndBufSize());
//                bootstrap.childOption(ChannelOption.SO_SNDBUF, this.nettyServerConfig.getServerSocketSndBufSize());
//            }
//
//            if (this.nettyServerConfig.getServerSocketRcvBufSize() > 0) {
//                logger.info("server set SO_RCVBUF to {}", this.nettyServerConfig.getServerSocketRcvBufSize());
//                bootstrap.childOption(ChannelOption.SO_RCVBUF, this.nettyServerConfig.getServerSocketRcvBufSize());
//            }
//
//            if (this.nettyServerConfig.getWriteBufferLowWaterMark() > 0 && this.nettyServerConfig.getWriteBufferHighWaterMark() > 0) {
//                logger.info("server set netty WRITE_BUFFER_WATER_MARK to {},{}", this.nettyServerConfig.getWriteBufferLowWaterMark(), this.nettyServerConfig.getWriteBufferHighWaterMark());
//                bootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(this.nettyServerConfig.getWriteBufferLowWaterMark(), this.nettyServerConfig.getWriteBufferHighWaterMark()));
//            }
//
//            if (this.nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
//                bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//            }

//        Channel channel = bootstrap.bind().sync().channel();
//        channel.closeFuture();

        try {
            ChannelFuture sync = this.serverBootstrap.bind().sync();
            logger.info("Started micro rpc server on port {}", nettyServerConfig.getPort());
        } catch (InterruptedException e) {
            throw new RpcServerException("this.serverBootstrap.bind().sync() InterruptedException", e);
        }
    }

    public void shutdown() {
        try {
            this.eventLoopGroupBoss.shutdownGracefully();
            this.eventLoopGroupSelector.shutdownGracefully();
        } catch (Exception e) {
            logger.error("NettyRemotingServer shutdown exception, ", e);
        }
    }

    public void stop() {
        // destroy server thread
        if (thread != null && thread.isAlive()) {
            thread.interrupt();
        }
    }

    private boolean useEpoll() {
        return RemotingUtil.isLinuxPlatform()
                && nettyServerConfig.isUseEpollNativeSelector()
                && Epoll.isAvailable();
    }

    @ChannelHandler.Sharable
    class NettyConnectManageHandler extends ChannelDuplexHandler {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY SERVER PIPELINE: channelRegistered {}", remoteAddress);
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY SERVER PIPELINE: channelUnregistered, the channel[{}]", remoteAddress);
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY SERVER PIPELINE: channelActive, the channel[{}]", remoteAddress);
            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            super.channelInactive(ctx);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    logger.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
                    RemotingUtil.closeChannel(ctx.channel());
                }
            }

            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.warn("NETTY SERVER PIPELINE: exceptionCaught {}", remoteAddress);
            logger.warn("NETTY SERVER PIPELINE: exceptionCaught exception.", cause);
            RemotingUtil.closeChannel(ctx.channel());
        }
    }
}
